package com.upcoming.data.storm.kafka;

import com.upcoming.data.storm.util.Event;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

/**
 * 作者：tianjiayuan
 * 创建时间：2018-05-23 14:44
 * 类描述：
 * 修改人：
 * 修改时间：
 */
public class EventBolt extends BaseRichBolt {

    private static final Logger logger = LoggerFactory.getLogger(EventBolt.class);

    private OutputCollector collector;

    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.collector = outputCollector;
    }

    public void execute(Tuple tuple) {
        Event event = (Event)tuple.getValue(0);
        logger.info(event.getEventMsg());
        this.collector.ack(tuple);
    }

    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }
}
